// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package src

import (
	"context"
	"fmt"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/runtime"
	"gitee.com/gomq/sr"
	amqp "github.com/rabbitmq/amqp091-go"
	"math/rand"
	"sync"
	"time"
)

const (
	defaultMessageTtl = 604800000 // 7 days (milliseconds)

	headerConsumedCount = "msr-consumed-count"
	headerDelaySeconds  = "msr-delay-seconds"
	headerProducedCount = "msr-produced-count"
	headerProducedTime  = "msr-produced-time"
)

type (
	// This is a component used to operate rabbitmq settings directly.
	client struct {
		manager *manager

		ac amqp.Config
		mu *sync.RWMutex
	}

	// This is a component used to define subscription definitions.
	binding struct {
		Queue sr.QueueName
		Topic sr.TopicName
	}
)

func (o *client) CloseChannel(ctx context.Context, channel *amqp.Channel) {
	if channel == nil || channel.IsClosed() {
		return
	}
	if err := channel.Close(); err != nil {
		log.Warnfc(ctx, `[servlet=rabbitmq][client] channel close: error="%v"`, err)
	} else if log.Config().DebugOn() {
		log.Debugfc(ctx, `[servlet=rabbitmq][client] channel closed`)
	}
}

func (o *client) CloseConnection(ctx context.Context, connection *amqp.Connection) {
	if connection == nil || connection.IsClosed() {
		return
	}
	if err := connection.Close(); err != nil {
		log.Warnfc(ctx, `[servlet=rabbitmq][client] connection close: addr="%s", error="%v"`, connection.RemoteAddr(), err)
	} else if log.Config().DebugOn() {
		log.Debugfc(ctx, `[servlet=rabbitmq][client] connection closed: addr="%s"`, connection.RemoteAddr())
	}
}

func (o *client) Consume(ctx context.Context, channel *amqp.Channel, queue sr.QueueName, prefetch int32) (list <-chan amqp.Delivery, err error) {
	if err = channel.Qos(int(prefetch), 0, false); err == nil {
		uid := fmt.Sprintf(`a%db%dc%d`, time.Now().Nanosecond(), rand.Int31n(9999), rand.Int31n(9999))
		list, err = channel.ConsumeWithContext(
			ctx,
			queue.String(),
			uid, false, false, false, false, nil,
		)
	}
	if err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] receiver create: queue="%s", qos="%d", error="%v"`, queue, prefetch, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] receiver created: queue="%s", qos="%d"`, queue, prefetch)
	}
	return
}

func (o *client) DeclareBinding(ctx context.Context, channel *amqp.Channel, queue sr.QueueName, topic sr.TopicName, tag sr.TopicTag) (err error) {
	if err = channel.QueueBind(
		queue.String(),
		tag.String(),
		topic.String(), false, nil,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] binding declare: queue="%s", topic="%s", tag="%s", error="%v"`, queue, topic, tag, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] binding declared: queue="%s", topic="%s", tag="%s"`, queue, topic, tag)
	}
	return
}

func (o *client) DeclareExchange(ctx context.Context, channel *amqp.Channel, topic sr.TopicName) (err error) {
	if err = channel.ExchangeDeclare(topic.String(),
		"topic", true, false, false, false, nil,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] exchange declare: exchange="%s", error="%v"`, topic, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] exchange declared: exchange="%s"`, topic)
	}
	return
}

func (o *client) DeclareQueue(ctx context.Context, channel *amqp.Channel, queue sr.QueueName, args amqp.Table) (err error) {
	if _, err = channel.QueueDeclare(
		queue.String(), true, false, false, false,
		args,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] queue declare: queue="%s", error="%v"`, queue, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] queue declared: queue="%s"`, queue)
	}
	return
}

func (o *client) DeleteBinding(ctx context.Context, channel *amqp.Channel, queue sr.QueueName, topic sr.TopicName, tag sr.TopicTag) (err error) {
	if err = channel.QueueUnbind(
		queue.String(),
		tag.String(),
		topic.String(), nil,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] binding delete: queue="%s", exchange="%s", error="%v"`, queue, topic, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] binding deleted: queue="%s", exchange="%s"`, queue, topic)
	}
	return
}

func (o *client) DeleteExchange(ctx context.Context, channel *amqp.Channel, topic sr.TopicName) (err error) {
	if err = channel.ExchangeDelete(
		topic.String(), false, false,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] exchange delete: exchange="%s", error="%v"`, topic, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] exchange deleted: exchange="%s"`, topic)
	}
	return
}

func (o *client) DeleteQueue(ctx context.Context, channel *amqp.Channel, queue sr.QueueName) (err error) {
	if _, err = channel.QueueDelete(
		queue.String(), false, false, false,
	); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] queue delete: queue="%s", error="%v"`, queue, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] queue deleted: queue="%s"`, queue)
	}
	return
}

// GenerateExchange generate a normal exchange and a delay exchange and a
// dead exchange for the task with given subscription id and topic name.
//
//	return "x.topic",
//	       "{delay}.x.topic.1001",
//	       "{dead}.x.topic.1001"
func (o *client) GenerateExchange(sid int, topic sr.TopicName) (exchange, delay, dead sr.TopicName) {
	exchange = sr.TopicName(fmt.Sprintf(`%s%s`, o.manager.config.Prefix, topic.ToLower())) // x.orders
	delay = sr.TopicName(fmt.Sprintf(`{delay}.%stopic.%d`, o.manager.config.Prefix, sid))  // {delay}.x.topic.1
	dead = sr.TopicName(fmt.Sprintf(`{dead}.%stopic.%d`, o.manager.config.Prefix, sid))    // {dead}.x.topic.1
	return
}

// GenerateMapping generate a mapping configurations for exchanges, queues
// and bindings.
func (o *client) GenerateMapping(subscription sr.Subscription) (exchanges []sr.TopicName, queues []map[sr.QueueName]amqp.Table, bindings []binding) {
	// Generate exchanges.
	ex, exDelay, exDead := o.GenerateExchange(subscription.GetId(), subscription.GetTopic())
	exchanges = []sr.TopicName{ex, exDead, exDelay}

	// Generate queues.
	qu, quDelay := o.GenerateQueue(subscription.GetId())
	queues = []map[sr.QueueName]amqp.Table{
		{qu: {"x-message-ttl": defaultMessageTtl}},
		{quDelay: {"x-message-ttl": defaultMessageTtl, "x-dead-letter-exchange": exDead.String()}},
	}

	// Generate bindings.
	bindings = []binding{
		{
			Queue: qu,
			Topic: ex,
		},
		{
			Queue: qu,
			Topic: exDead,
		},
		{
			Queue: quDelay,
			Topic: exDelay,
		},
	}
	return
}

func (o *client) GeneratePublishing() amqp.Publishing {
	publishing := amqp.Publishing{
		Headers:         make(amqp.Table),
		ContentType:     "text/plain",
		ContentEncoding: "utf-8",
		DeliveryMode:    amqp.Persistent,
	}
	return publishing
}

func (o *client) GeneratePublishingWithPayload(payload *sr.Payload) amqp.Publishing {
	// Initial
	// publishing attributes.
	publishing := o.GeneratePublishing()
	publishing.Body = payload.Body
	publishing.MessageId = payload.Hash.String()
	publishing.Timestamp = payload.Time

	// Expiration supported.
	if payload.DelaySeconds > 0 {
		publishing.Headers[headerDelaySeconds] = payload.DelaySeconds * 1000
	}

	// Attach counter.
	publishing.Headers[headerConsumedCount] = 0
	publishing.Headers[headerProducedCount] = 1
	publishing.Headers[headerProducedTime] = payload.Time.UnixMicro()
	return publishing
}

// GenerateQueue generate a normal queue and a delay queue for the task with
// given subscription id.
//
//	return "x.queue.1001",
//	       "{delay}.x.queue.1001"
func (o *client) GenerateQueue(sid int) (queue, delay sr.QueueName) {
	queue = sr.QueueName(fmt.Sprintf(`%squeue.%d`, o.manager.config.Prefix, sid))         // x.queue.1
	delay = sr.QueueName(fmt.Sprintf(`{delay}.%squeue.%d`, o.manager.config.Prefix, sid)) // {delay}.x.queue.1
	return
}

func (o *client) OpenChannel(ctx context.Context, connection *amqp.Connection) (channel *amqp.Channel, err error) {
	if channel, err = connection.Channel(); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] channel open: error="%v"`, err)
	} else if log.Config().DebugOn() {
		log.Debugfc(ctx, `[servlet=rabbitmq][client] channel opened`)
	}
	return
}

func (o *client) OpenConnection(ctx context.Context) (connection *amqp.Connection, err error) {
	if connection, err = amqp.DialConfig(o.manager.config.Url, o.ac); err != nil {
		log.Errorfc(ctx, `[servlet=rabbitmq][client] connection open: error="%v"`, err)
	} else {
		log.Infofc(ctx, `[servlet=rabbitmq][client] connection opened: addr="%s"`, connection.RemoteAddr())
	}
	return
}

func (o *client) Publish(ctx context.Context, channel *amqp.Channel, topic sr.TopicName, tag sr.TopicTag, msg amqp.Publishing) (err error) {
	if err = channel.PublishWithContext(
		ctx,
		topic.String(),
		tag.String(), false, false,
		msg,
	); err != nil {
		runtime.GetCounter().MqCounter().IncrFailProducedCount(1)
		log.Errorfc(ctx, `[servlet=rabbitmq][client] publish failed: topic="%s", tag="%s", expiration="%s", error="%v"`, topic, tag, msg.Expiration, err)
	} else {
		runtime.GetCounter().MqCounter().IncrSuccessProducedCount(1)
		log.Infofc(ctx, `[servlet=rabbitmq][client] publish succeed: topic="%s", tag="%s", expiration="%s", message-id="%s"`, topic, tag, msg.Expiration, msg.MessageId)
	}
	return
}

// + Access

func (o *client) init() *client {
	o.mu = &sync.RWMutex{}

	// Prepare amqp configuration used to create rabbitmq connection.
	o.ac = amqp.Config{Vhost: o.manager.config.Host}

	// Add authentication into connection configuration.
	if o.manager.config.Username != "" && o.manager.config.Password != "" {
		o.ac.SASL = []amqp.Authentication{
			&amqp.AMQPlainAuth{
				Username: o.manager.config.Username,
				Password: o.manager.config.Password,
			},
		}
	}
	return o
}
